#include <time.h>
#include <math.h>
#include <stdio.h>
#include <stdbool.h>
#include <string.h>
#include <inttypes.h>
#include <st.h>
#define USE_LOCAL_LOG_LEVEL
#include "log.h"
#include "ursax.h"
#include "statistics.h"

#ifndef LIB_VOLUME_SO
#include "backend-fs.h"
#include "connection-pool.h"
#include "netserver.h"
#else
#include "concurrent-connection-pool.h"
#endif

#define URSAX_STATIS_ITEM_MAX (7)
#define RB_MAX_LENGTH (900)
#define URSAX_STATIS_REPORT_MAX (4)

typedef struct ring_buffer {
	int front;
	int rear;
	int length;
	uint32_t *data;
	ring_buffer(){
		memset(this, 0, sizeof(*this));
	}
	ring_buffer(int len){
		front = 0;
		rear = 0;
		length = len+2;
		data = new uint32_t[len+2];
	}
	~ring_buffer(){
		delete[] data;
		data = NULL;
	}
} ring_buffer;

struct ursax_statis_reporter {
	const char *name;
	int interval;
	int cnt;
	uint64_t avg_sum;
	uint64_t square_sum;
	ursax_statis_reporter(const char* name, int interval) {
		memset(this, 0, sizeof(*this));
		this->name = name;
		this->interval = interval;
	}
};

typedef void (*set_value_t) (ursax_statis_types, int);
typedef void (*report_t) (ursax_statis_types, char *, int);

struct ursax_statis {
	const char *item_name;
    ring_buffer sampling;
    uint32_t sample_per_s; /**< sample value within 1 second.*/
    int cnt_per_s; /**< sample count with 1 second.*/
    ursax_statis_reporter *reporters;
    time_t last_inc_time;
    set_value_t set_value;
    report_t report;
    ursax_statis(){
    	memset(this, 0, sizeof(*this));
    }
    ursax_statis(const char* item_name, int ringbuffer_len, set_value_t set_value, report_t report) : sampling(ringbuffer_len) {
    	this->item_name = item_name;
    	this->last_inc_time = 0;
    	this->sample_per_s = 0;
    	this->cnt_per_s = 0;
    	this->set_value = set_value;
    	this->report = report;
    	this->reporters = new ursax_statis_reporter[URSAX_STATIS_REPORT_MAX]{{"1s", 1}, {"1m", 60}, {"5m", 300}, {"15m", 900}};
    }
    ~ursax_statis(){
    	delete[] reporters;
		reporters = NULL;
    }
};

static ursax_statis *statis_data[URSAX_STATIS_ITEM_MAX];
static st_thread_t statis_th;
static bool do_run;

static bool is_queue_empty(ring_buffer const *rbuffer)
{
	return (rbuffer->front == rbuffer->rear);
}

static bool is_queue_full(ring_buffer const *rbuffer)
{
	return ((rbuffer->rear+1) % rbuffer->length == rbuffer->front);
}

/** keep it for development debug purpose.
static int
get_queue_length(ring_buffer const *rbuffer)
{
	return (rbuffer->rear - rbuffer->front + rbuffer->length) % rbuffer->length;
}
*/

static int enqueue(ring_buffer *rbuffer, bool increase, int value)
{
	int idx = 0;

	if (is_queue_full(rbuffer) && increase) {
		return -1;
	}

	if (is_queue_empty(rbuffer) && !increase) {
		return -2;
	}

	idx = increase ? rbuffer->rear : ((rbuffer->rear-1+rbuffer->length) % rbuffer->length);

	if (increase) {
		if (0 <= value) {
			rbuffer->data[idx] = value;
			rbuffer->rear = ((idx + 1) % rbuffer->length);
		} /**< ignore negative value set to avoid huge uint_32.*/
	} else {
		if ((0 > value) && (rbuffer->data[idx] < (uint32_t)(-1*value))) {
			return 0;
		}
		rbuffer->data[idx] += value;
	}

	return 0;
}

static int dequeue(ring_buffer *rbuffer, uint32_t *value)
{
	if (is_queue_empty(rbuffer)) {
		return -1;
	}

	if (NULL != value) {
		*value = rbuffer->data[rbuffer->front];
	}
	rbuffer->front = ((rbuffer->front+1) % rbuffer->length);

	return 0;
}

static int get_queue_idx_by_distance_rear(ring_buffer const *rbuffer, int distance)
{
	int rear = rbuffer->rear;
	int length = rbuffer->length;

	/**
	if (get_queue_length(rbuffer) < distance) {
		return rbuffer->front;
	} else {
		return (rear - distance + length) % length;
	}
	*/
	return (rear - 1 - distance + length) % length;
}

static bool is_index_in_queue(int index, ring_buffer const *rbuffer)
{
	int front = rbuffer->front;
	int rear = rbuffer->rear;
	int len = rbuffer->length;

	if (is_queue_empty(rbuffer)) {
		return false;
	} else if (front < rear) {
		return ((front <= index) && (index < rear));
	} else {
		return ((front <= index && index <= len) || (0 <= index && index < rear));
	}
}

static int get_queue_rear_data(ring_buffer const *rbuffer, uint32_t *value)
{
	int idx = 0; 
	if (is_queue_empty(rbuffer)) {
		return -1;
	}

	idx = (rbuffer->rear-1 + rbuffer->length) % rbuffer->length;
	*value = rbuffer->data[idx];
	return 0;
}

/**
 * keep update_queue_rear_data() and _get_queue_content() for development debug purpose.
static int
update_queue_rear_data(ring_buffer *rbuffer, uint32_t value)
{
	int idx = 0; 
	if (is_queue_empty(rbuffer)) {
		return -1;
	}

	idx = (rbuffer->rear-1 + rbuffer->length) % rbuffer->length;
	rbuffer->data[idx] = value;
	return 0;
}

static char *
_get_queue_content(ring_buffer const *rbuffer)
{
    static char content[8192];
    int idx = rbuffer->front;
    char *ptr = content;
    int len = sizeof(content);

    memset(content, sizeof(content), '\0');
    ptr += snprintf(ptr, len, "QUEUE CONTENT:Front:%d, Rear:%d Length: %d [",
				rbuffer->front, rbuffer->rear, get_queue_length(rbuffer));
    while (idx != rbuffer->rear) {
        ptr += snprintf(ptr, len - (ptr-content), "%u, ", rbuffer->data[idx]);
        idx = (idx+1) % rbuffer->length;
    }
    snprintf(ptr, len - (ptr-content), "]");

    return content;
}

// to show queue content use: LOG_DEBUG(_get_queue_content(XXX))
*/

#ifndef LIB_VOLUME_SO
extern struct NetServer *g_chunk_ns;
#endif
static uint32_t ursax_statis_get_sample_val(ursax_statis_types type)
{
	uint32_t value = 0;
	uint32_t sample = 0;
	int count = 0;
	ursax_statis *item = statis_data[type];

	switch (type) {
#ifndef LIB_VOLUME_SO
		case FD_NUMBER:
			value = get_fd_number();
			break;
		case IN_CONNECT:
			value = server_get_nconnections(g_chunk_ns);
#endif
		case OUT_CONNECT:
#ifndef LIB_VOLUME_SO
			value = get_connection_number();
#else
			value = get_connection_number();
#endif
			break;
		default:
			sample = item->sample_per_s;
			count = item->cnt_per_s == 0 ? 1 : item->cnt_per_s;
			value = sample / count;
			break;
	}

	return value;
}

/**< this function will run in a seprated thread and do statis processing.*/
static void ursax_statis_processor()
{
	int type = IOPS;
	for (type = IOPS; type < TYPE_MAX; type++) {
		ursax_statis *item = statis_data[type];
		ring_buffer *rb = &item->sampling;

		/**< accumulate value to reporters in type */
		int rptor_idx = 0;
		//uint32_t sample = item->sample_per_s;
		//int count = item->cnt_per_s == 0 ? 1 : item->cnt_per_s;
		uint32_t val = ursax_statis_get_sample_val((ursax_statis_types)type);
		for (rptor_idx = 0; rptor_idx < URSAX_STATIS_REPORT_MAX; rptor_idx++) {
				item->reporters[rptor_idx].avg_sum += (uint64_t)val;
				item->reporters[rptor_idx].square_sum += ((uint64_t)val * (uint64_t)val);
				item->reporters[rptor_idx].cnt += 1;

				int start = get_queue_idx_by_distance_rear(rb, item->reporters[rptor_idx].interval-1);
				if (is_index_in_queue(start, rb)) {
					uint32_t val_del = rb->data[start];
					item->reporters[rptor_idx].avg_sum -= (uint64_t)val_del;
					item->reporters[rptor_idx].square_sum -= ((uint64_t)val_del * (uint64_t)val_del);
					item->reporters[rptor_idx].cnt -= 1;
				}
		}

		/**< store sample value in sampling array in type */
		int ret = 0;
		int tried = 0;
		while (0 != (ret = enqueue(rb, true, val)) && (tried++ < 2)) {
			if (-1 == ret) {
				/**< queue is full. keep it for development debug purpose.
				LOG_DEBUG("[%s:%d] %s, BUFFER IS FULL. front:%d, rear:%d",
							__func__, __LINE__, item->item_name, item->sampling.front, item->sampling.rear);
				*/
				dequeue(&item->sampling, NULL);
			} else {
				LOG_ERROR("[%s:%d] Error: update an empty sampling data queue.",
							__func__, __LINE__);
				break;
			}
		}

		/**< 3. clear cnt_per_s and sample_per_s */
		item->sample_per_s = 0;
		item->cnt_per_s = 0;
	}
}

/**
 * statis thread.
 */
static bool thread_in_sleep = false;
static void *ursax_statis_processor_thread(void *arg)
{
	while (do_run) {
		LOG_DEBUG("ursax_statis_processor start");
		ursax_statis_processor();
		LOG_DEBUG("ursax_statis_processor end");
		thread_in_sleep = true;
		st_sleep(10);
		thread_in_sleep = false;
	}

	LOG_INFO("ursax_statis_processor_thread exit");
	return NULL;
}

int ursax_statis_init(void)
{
//	register_local_log_level("c.statistics");

    return 0;

	if(statis_th) {
		LOG_DEBUG("statistics start already");
		return 0;
	}

	char msg[128];
	int ret = 0;
	do_run = true;
	statis_th = st_thread_create(ursax_statis_processor_thread, NULL, 1, ST_STACK_SIZE);

	if (NULL == statis_th) {
		snprintf(msg, sizeof(msg), "failed, errno:[%d], msg:[%s]", errno, strerror(errno));
		ret = -1;
	} else {
		snprintf(msg, sizeof(msg), "successfully");
	}

	LOG_INFO("[%s:%d] starting ursax_statis_processor_thread [%lx] %s", __func__, __LINE__, statis_th, msg);
	st_usleep(1000);

	return ret;
}

void ursax_statis_fini(void)
{
	if(do_run) {
		do_run = false;
		if(thread_in_sleep){
			st_thread_interrupt(statis_th);
		}
		st_thread_join(statis_th, NULL);
	}
}

static void ursax_statis_report_common(ursax_statis_types type, char *cmd, int cmd_len)
{
	ursax_statis *item = statis_data[type];
	int reporter_idx = 0;
	int cnt = 0;
	char *cmd_ptr = cmd;
	int len = cmd_len;
	int num = 0;

	if (NULL == cmd) {
		return;
	}

	cnt = snprintf(cmd_ptr, len, " %s ", item->item_name);
	for (reporter_idx = 0; reporter_idx < URSAX_STATIS_REPORT_MAX; reporter_idx++) {
		struct ursax_statis_reporter *reporter = &item->reporters[reporter_idx];
		num = reporter->cnt == 0 ? 1 : reporter->cnt;
		double avg = (double)reporter->avg_sum / num;
		double std_devi = (double)reporter->square_sum / num;
		std_devi = sqrt(std_devi - (avg * avg));
		/** keep it for development debug purpose.
		LOG_DEBUG("[%s:%d] [%s:%s] num:%d, report->avg_sum:%lu, reporter->square_sum:%lu, avg: %f, std_devi:%f",
					__func__, __LINE__, item->item_name, reporter->name, num,
					reporter->avg_sum, reporter->square_sum, avg, std_devi);
		*/

		cmd_ptr += cnt;
		len -= cnt;
		cnt = snprintf(cmd_ptr, len, "%d,%d,", (int)avg, (int)std_devi);
	}
	cmd_ptr[cnt-1] = '\0';
}

static inline void ursax_statis_set_refer_last(ursax_statis_types type, int value)
{
	int value_to_set = 0;

	if (statis_data[type]->sample_per_s == 0) {
		ring_buffer *rb = &statis_data[type]->sampling;
		uint32_t last_val = 0;
		get_queue_rear_data(rb, &last_val);
		value_to_set = (long int)last_val + value;
	} else {
		value_to_set = value;
	}

	if (value_to_set < 0 && statis_data[type]->sample_per_s < (uint32_t)(-1*value_to_set) ) {
		value_to_set = 0;
	}

	statis_data[type]->sample_per_s += value_to_set;
}

static inline void ursax_statis_set_direct(ursax_statis_types type, int value)
{
	statis_data[type]->sample_per_s += value;
}

static inline void ursax_statis_set_counted(ursax_statis_types type, int value)
{
	statis_data[type]->cnt_per_s += 1;
	statis_data[type]->sample_per_s += value;
}

__attribute__ ((__constructor__))
static void init_data()
{
	statis_data[IOPS] = new ursax_statis("iops", RB_MAX_LENGTH, ursax_statis_set_direct, ursax_statis_report_common);
	statis_data[THROUGHPUT] = new ursax_statis("throughput", RB_MAX_LENGTH, ursax_statis_set_direct, ursax_statis_report_common);
	statis_data[LATENCY] = new ursax_statis("latency", RB_MAX_LENGTH, ursax_statis_set_counted, ursax_statis_report_common);
	statis_data[QUEUE_LENGTH] = new ursax_statis("queue", RB_MAX_LENGTH, ursax_statis_set_refer_last, ursax_statis_report_common);
#ifndef LIB_VOLUME_SO
	statis_data[FD_NUMBER] = new ursax_statis("fd_number", RB_MAX_LENGTH, NULL, ursax_statis_report_common);
	statis_data[IN_CONNECT] = new ursax_statis("in_conn", RB_MAX_LENGTH, ursax_statis_set_refer_last, ursax_statis_report_common);
#endif
	statis_data[OUT_CONNECT] = new ursax_statis("out_conn", RB_MAX_LENGTH, NULL, ursax_statis_report_common);
}

void ursax_statis_set_value(ursax_statis_types type, int value)
{
	statis_data[type]->set_value(type, value);
}

uint64_t
ursax_statis_begin_io(void)
{
	struct timeval ts;
	ursax_statis_set_value(QUEUE_LENGTH, 1);
	gettimeofday(&ts, NULL);

	return (uint64_t)(ts.tv_sec * 1000000 + ts.tv_usec);
}

void
ursax_statis_end_io(uint64_t us, uint32_t io_size_kb)
{
	struct timeval te;
	uint64_t elapsed;

	gettimeofday(&te, NULL);
	elapsed = (uint64_t)(te.tv_sec * 1000000 + te.tv_usec) - us;

	ursax_statis_set_value(LATENCY, elapsed);
	ursax_statis_set_value(IOPS, 1);
	ursax_statis_set_value(QUEUE_LENGTH, -1);
	ursax_statis_set_value(THROUGHPUT, io_size_kb);
}

void
ursax_statis_report(ursax_statis_types type, char *cmd_buf, int cmd_len)
{
	if (NULL == cmd_buf || 0 > cmd_len) {
		LOG_ERROR("[%s:%d] invalid arguments: entity or identity is NULL.",
				__func__, __LINE__);
		return;
	}
	statis_data[type]->report(type, cmd_buf, cmd_len);
}

void ursax_statis_report_all(char *cmd_buf, int cmd_len)
{
	int type;
	char *cmd_ptr = NULL;
	int len = 0;
	int cnt = 0;

	if (NULL == cmd_buf || 0 >= cmd_len) {
		LOG_ERROR("[%s:%d] invalid arguments: cmd_buf is NULL or cmd_len is non-positive.",
					__func__, __LINE__);
		return;
	}

	for (type = IOPS; type < TYPE_MAX; type++) {
		cmd_ptr = cmd_buf + cnt;
		len = cmd_len - cnt;
		if (0 >= len) {
			LOG_ERROR("[%s:%d] invalid arguments: cmd_buf is too small.",
						__func__, __LINE__);
		}

		ursax_statis_report((ursax_statis_types)type, cmd_ptr, len);
		cnt = strlen(cmd_buf);
	}
	/**< add a timestamp at the end */
	time_t ts = time(NULL);
	cnt += strftime(cmd_buf+cnt, cmd_len-cnt, " timestamp %Y-%m-%d-%H-%M-%S", localtime(&ts));

	int major_version, minor_version, patch_version;
	get_version_nbs(&major_version, &minor_version, &patch_version);
	snprintf(cmd_buf+cnt, cmd_len-cnt, " client-version %d-%d-%d ", major_version, minor_version, patch_version);
}

void statis_set_value(ursax_statis_types type, int value)
{
    statis_data[type]->set_value(type, value);
}

uint64_t
statis_begin_io(void)
{
	struct timeval ts;
    statis_set_value(QUEUE_LENGTH, 1);
	gettimeofday(&ts, NULL);

	return (uint64_t)(ts.tv_sec * 1000000 + ts.tv_usec);
}

void
statis_end_io(uint64_t us, uint32_t io_size_kb)
{
	struct timeval te;
	uint64_t elapsed;

	gettimeofday(&te, NULL);
	elapsed = (uint64_t)(te.tv_sec * 1000000 + te.tv_usec) - us;

    statis_set_value(LATENCY, elapsed);
    statis_set_value(IOPS, 1);
    statis_set_value(QUEUE_LENGTH, -1);
    statis_set_value(THROUGHPUT, io_size_kb);
}